#[tokio::test]
async fn test_thunderbroker_prepare_request_with_event_handler() {
    // Create a BrokerRequest with an event_handler
    let alias = "TestPlugin.onTestEvent";
    let event_handler_method = "process_event_handler_method";

    let rpc_request = BrokerRequest {
        rpc: RpcRequest::get_new("TestPlugin.onTestEvent".to_owned(), None),
        rule: Rule {
            alias: alias.to_owned(),
            transform: RuleTransform::default(),
            endpoint: None,
            filter: None,
            event_handler: Some(event_handler_method.to_owned()),
            sources: None,
        },
        subscription_processed: None,
        workflow_callback: None,
    };

    // Create a ThunderBroker instance (mocking necessary dependencies)
    let (tx, _rx) = mpsc::channel(1);
    let (sender, _rec) = mpsc::channel(1);
    let send_data = vec![];

    let thndr_broker = get_thunderbroker(tx, send_data, sender.clone(), false).await;

    // Prepare the request
    let prepared_requests = thndr_broker.prepare_request(&rpc_request);
    assert!(
        prepared_requests.is_ok(),
        "Failed to prepare request with event_handler: {:?}",
        prepared_requests
    );

    let requests = prepared_requests.unwrap();

    // Verify that the request contains the correct method and handling
    for request in requests {
        assert!(
            request.contains(&format!("\"method\":\"{}\"", rpc_request.rule.alias)),
            "Prepared request does not contain the correct method: {}",
            request
        );
    }
}

#[tokio::test]
async fn test_thunderbroker_handle_event_with_event_handler() {
    // Set up the channels and the sender for the test
    let (tx, _rx) = mpsc::channel(1);
    let (sender, mut rec) = mpsc::channel(1);

    // Create mock data to simulate the websocket response
    let mock_response = json!({
        "jsonrpc": "2.0",
        "method": "TestPlugin.onTestEvent",
        "params": {
            "data": "event_data"
        }
    })
    .to_string();

    let send_data = vec![WSMockData::get(mock_response.clone())];

    // Initialize the ThunderBroker
    let thndr_broker = get_thunderbroker(tx, send_data, sender.clone(), false).await;

    // Register a custom callback to verify event handling
    let callback_id = 12345;
    let custom_callback = BrokerCallback {
        sender: sender.clone(),
    };
    thndr_broker
        .register_custom_callback(callback_id, custom_callback)
        .await;

    // Simulate receiving an event
    ThunderBroker::handle_jsonrpc_response(
        mock_response.as_bytes(),
        thndr_broker.get_default_callback(),
    );

    // Verify that the event handler was invoked correctly
    let v = tokio::time::timeout(Duration::from_secs(2), rec.recv())
        .await
        .expect("Timeout while waiting for response");

    if let Some(broker_output) = v {
        let data = broker_output
            .data
            .result
            .expect("No result in response data");
        let data_value = data.get("data").expect("Data not found in response");
        let data_str = data_value.as_str().expect("Value is not a string");
        assert_eq!(data_str, "event_data");
    } else {
        panic!("Received None instead of a valid response");
    }

    // Clean up by unregistering the custom callback
    thndr_broker.unregister_custom_callback(callback_id).await;
}

#[tokio::test]
async fn test_thunderbroker_subscribe_with_event_handler() {
    let (tx, _rx) = mpsc::channel(1);
    let (sender, _rec) = mpsc::channel(1);
    let send_data = vec![];

    let thndr_broker = get_thunderbroker(tx, send_data, sender.clone(), false).await;

    // Create a subscription request with event_handler
    let alias = "TestPlugin.onTestEvent";
    let event_handler_method = "process_event";

    let subscribe_request = BrokerRequest {
        rpc: RpcRequest::get_new_internal(
            alias.to_owned(),
            Some(json!({"listen": true})),
        ),
        rule: Rule {
            alias: alias.to_owned(),
            transform: RuleTransform::default(),
            endpoint: None,
            filter: None,
            event_handler: Some(event_handler_method.to_owned()),
            sources: None,
        },
        subscription_processed: Some(false),
        workflow_callback: None,
    };

    // Subscribe to an event
    let unsubscribe = thndr_broker.subscribe(&subscribe_request);

    // Verify that unsubscribe is None since this is a new subscription
    assert!(
        unsubscribe.is_none(),
        "Expected no prior subscription to unsubscribe"
    );

    // Now, subscribe again to the same event, which should return the previous subscription to be unsubscribed
    let unsubscribe = thndr_broker.subscribe(&subscribe_request);

    // Verify that the previous subscription is returned for unsubscription
    assert!(
        unsubscribe.is_some(),
        "Expected previous subscription to be returned for unsubscription"
    );

    // Unsubscribe
    if let Some(unsubscribe_request) = unsubscribe {
        // Here you can test handling the unsubscription if needed
        // For example, ensure that the unsubscribe request is correctly formatted
        assert_eq!(
            unsubscribe_request.rpc.method,
            subscribe_request.rpc.method
        );
    }
}

#[tokio::test]
async fn test_thunderbroker_check_and_generate_plugin_activation_request_with_event_handler() {
    // Create a BrokerRequest with an event_handler
    let alias = "SomePlugin.someMethod";
    let event_handler_method = "process_event";

    let rpc_request = BrokerRequest {
        rpc: RpcRequest::get_new("SomePlugin.someMethod".to_owned(), None),
        rule: Rule {
            alias: alias.to_owned(),
            transform: RuleTransform::default(),
            endpoint: None,
            filter: None,
            event_handler: Some(event_handler_method.to_owned()),
            sources: None,
        },
        subscription_processed: None,
        workflow_callback: None,
    };

    // Create a ThunderBroker instance
    let (tx, _rx) = mpsc::channel(1);
    let (sender, _rec) = mpsc::channel(1);
    let send_data = vec![];

    let thndr_broker = get_thunderbroker(tx, send_data, sender.clone(), false).await;

    // Test plugin activation request generation
    let result = thndr_broker.check_and_generate_plugin_activation_request(&rpc_request);

    // Since the StatusManager may not have the plugin status, it should generate a plugin status request
    assert!(
        result.is_ok(),
        "Failed to generate plugin activation request: {:?}",
        result
    );

    let requests = result.unwrap();
    assert!(
        !requests.is_empty(),
        "Expected plugin activation requests to be generated."
    );
}

// Add these tests to your `thunder_broker.rs` file or a separate test module.

#[cfg(test)]
mod tests {
    use super::*;
    use ripple_sdk::tokio::sync::mpsc::channel;
    use ripple_sdk::tokio;
    use ripple_sdk::api::gateway::rpc_gateway_api::{RpcRequest, ApiProtocol, CallContext};
    use ripple_sdk::utils::error::RippleError;
    use crate::state::platform_state::PlatformState;
    use crate::state::bootstrap_state::ChannelsState;
    use crate::test_utils::MockRippleClient;

    #[tokio::test]
    async fn test_get_broker() {
        // Arrange
        let (tx, _) = channel(2);
        let callback = BrokerCallback { sender: tx };
        let request = BrokerConnectRequest::default();
        let mut endpoint_broker_state = EndpointBrokerState::default();

        // Act
        let thunder_broker = ThunderBroker::get_broker(request.clone(), callback.clone(), &mut endpoint_broker_state);

        // Assert
        assert!(thunder_broker.ws_port > 0);
        assert_eq!(thunder_broker.callback.sender.capacity(), callback.sender.capacity());
    }

    #[tokio::test]
    async fn test_prepare_request_success() {
        // Arrange
        let (tx, _) = channel(2);
        let callback = BrokerCallback { sender: tx };
        let request = BrokerConnectRequest::default();
        let mut endpoint_broker_state = EndpointBrokerState::default();
        let thunder_broker = ThunderBroker::get_broker(request.clone(), callback.clone(), &mut endpoint_broker_state);

        let call_context = CallContext::new(
            "session_id".into(),
            "request_id".into(),
            "app_id".into(),
            1,
            ApiProtocol::Extn,
            "test_method".into(),
            Some("cid".into()),
            true,
        );
        let rpc_request = BrokerRequest::new(
            &RpcRequest::new("test_method".into(), "".into(), call_context),
            Rule::default(),
            None,
        );

        // Act
        let result = thunder_broker.prepare_request(&rpc_request);

        // Assert
        assert!(result.is_ok());
        let requests = result.unwrap();
        assert!(!requests.is_empty());
    }

    #[tokio::test]
    async fn test_prepare_request_failure() {
        // Arrange
        let (tx, _) = channel(2);
        let callback = BrokerCallback { sender: tx };
        let request = BrokerConnectRequest::default();
        let mut endpoint_broker_state = EndpointBrokerState::default();
        let thunder_broker = ThunderBroker::get_broker(request.clone(), callback.clone(), &mut endpoint_broker_state);

        let call_context = CallContext::new(
            "session_id".into(),
            "request_id".into(),
            "app_id".into(),
            1,
            ApiProtocol::Extn,
            "test_method".into(),
            Some("cid".into()),
            true,
        );
        let mut rpc_request = BrokerRequest::new(
            &RpcRequest::new("test_method".into(), "".into(), call_context),
            Rule::default(),
            None,
        );
        // Simulate an error condition if necessary

        // Act
        let result = thunder_broker.prepare_request(&rpc_request);

        // Assert
        assert!(result.is_ok()); // Adjust accordingly if you expect an error
    }

    #[tokio::test]
    async fn test_handle_jsonrpc_response_success() {
        // Arrange
        let (tx, mut rx) = channel(2);
        let callback = BrokerCallback { sender: tx };
        let response = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"success\":true}}";

        // Act
        ThunderBroker::handle_jsonrpc_response(response, callback.clone());
        let output = rx.recv().await.unwrap();

        // Assert
        assert!(output.data.result.is_some());
        assert_eq!(output.data.result.unwrap(), serde_json::json!({"success": true}));
    }

    #[tokio::test]
    async fn test_handle_jsonrpc_response_error() {
        // Arrange
        let (tx, mut rx) = channel(2);
        let callback = BrokerCallback { sender: tx };
        let response = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"code\": -32601, \"message\": \"Method not found\"}}";

        // Act
        ThunderBroker::handle_jsonrpc_response(response, callback.clone());
        let output = rx.recv().await.unwrap();

        // Assert
        assert!(output.data.error.is_some());
        let error = output.data.error.unwrap();
        assert_eq!(error.get("code").unwrap(), -32601);
        assert_eq!(error.get("message").unwrap(), "Method not found");
    }

    #[tokio::test]
    async fn test_get_cleaner() {
        // Arrange
        let (tx, _) = channel(2);
        let callback = BrokerCallback { sender: tx };
        let request = BrokerConnectRequest::default();
        let mut endpoint_broker_state = EndpointBrokerState::default();
        let thunder_broker = ThunderBroker::get_broker(request.clone(), callback.clone(), &mut endpoint_broker_state);

        // Act
        let cleaner = thunder_broker.get_cleaner();

        // Assert
        assert!(cleaner.cleaner.is_some());
    }

    #[tokio::test]
    async fn test_cleanup_session() {
        // Arrange
        let (cleaner_tx, mut cleaner_rx) = channel(2);
        let cleaner = BrokerCleaner {
            cleaner: Some(cleaner_tx),
        };

        // Act
        cleaner.cleanup_session("test_app").await;

        // Assert
        let app_id = cleaner_rx.recv().await.unwrap();
        assert_eq!(app_id, "test_app");
    }

    // Additional tests can be written to cover more functionalities as needed.
}

// Add these tests to your `thunder_broker.rs` file or a separate test module.

#[cfg(test)]
mod tests {
    use super::*;
    use ripple_sdk::api::gateway::rpc_gateway_api::{RpcRequest, ApiProtocol, CallContext};
    use ripple_sdk::tokio::sync::mpsc::{channel, Sender};
    use ripple_sdk::tokio;
    use ripple_sdk::utils::error::RippleError;
    use crate::broker::endpoint_broker::{BrokerCallback, BrokerConnectRequest, EndpointBrokerState, BrokerRequest};
    use crate::broker::rules_engine::{Rule, RuleTransform};
    use crate::state::platform_state::PlatformState;
    use crate::state::bootstrap_state::ChannelsState;
    use crate::service::extn::ripple_client::RippleClient;

    #[tokio::test]
    async fn test_thunder_broker_initialization() {
        // Arrange
        let (tx, _rx) = channel(1);
        let callback = BrokerCallback { sender: tx };
        let request = BrokerConnectRequest::default();
        let mut endpoint_broker_state = EndpointBrokerState::default();

        // Act
        let thunder_broker = ThunderBroker::get_broker(request.clone(), callback.clone(), &mut endpoint_broker_state);

        // Assert
        assert!(thunder_broker.ws_port > 0);
        assert_eq!(thunder_broker.callback.sender.capacity(), callback.sender.capacity());
    }

    #[tokio::test]
    async fn test_subscribe_success() {
        // Arrange
        let (tx, _rx) = channel(1);
        let callback = BrokerCallback { sender: tx };
        let mut thunder_broker = ThunderBroker::get_broker(BrokerConnectRequest::default(), callback.clone(), &mut EndpointBrokerState::default());
        let broker_request = BrokerRequest::default(); // Adjust as needed
        let subscription_id = "test_subscription".to_string();

        // Act
        let result = thunder_broker.subscribe(subscription_id.clone(), broker_request.clone()).await;

        // Assert
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_unsubscribe_success() {
        // Arrange
        let (tx, _rx) = channel(1);
        let callback = BrokerCallback { sender: tx };
        let mut thunder_broker = ThunderBroker::get_broker(BrokerConnectRequest::default(), callback.clone(), &mut EndpointBrokerState::default());
        let broker_request = BrokerRequest::default(); // Adjust as needed
        let subscription_id = "test_subscription".to_string();

        // First, subscribe
        let _ = thunder_broker.subscribe(subscription_id.clone(), broker_request.clone()).await;

        // Act
        let result = thunder_broker.unsubscribe(subscription_id.clone(), broker_request.clone()).await;

        // Assert
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_handle_jsonrpc_response_success() {
        // Arrange
        let response = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"success\":true}}";
        let (tx, mut rx) = channel(1);
        let callback = BrokerCallback { sender: tx };

        // Act
        ThunderBroker::handle_jsonrpc_response(response, callback.clone());

        // Assert
        if let Some(output) = rx.recv().await {
            assert!(output.data.result.is_some());
            assert_eq!(output.data.result.unwrap(), serde_json::json!({"success":true}));
        } else {
            panic!("Did not receive a response");
        }
    }

    #[tokio::test]
    async fn test_handle_jsonrpc_response_error() {
        // Arrange
        let response = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"code\":-32601,\"message\":\"Method not found\"}}";
        let (tx, mut rx) = channel(1);
        let callback = BrokerCallback { sender: tx };

        // Act
        ThunderBroker::handle_jsonrpc_response(response, callback.clone());

        // Assert
        if let Some(output) = rx.recv().await {
            assert!(output.data.error.is_some());
            let error = output.data.error.unwrap();
            assert_eq!(error.get("code").unwrap(), -32601);
            assert_eq!(error.get("message").unwrap(), "Method not found");
        } else {
            panic!("Did not receive a response");
        }
    }

    #[tokio::test]
    async fn test_prepare_request() {
        // Arrange
        let thunder_broker = ThunderBroker::get_broker(BrokerConnectRequest::default(), BrokerCallback::default(), &mut EndpointBrokerState::default());
        let call_context = CallContext::new(
            "session_id".to_string(),
            "request_id".to_string(),
            "app_id".to_string(),
            1,
            ApiProtocol::Extn,
            "test.method".to_string(),
            None,
            false,
        );

        let broker_request = BrokerRequest {
            rpc: RpcRequest::new("test.method".to_string(), "{}".to_string(), call_context),
            rule: Rule {
                alias: "test.method".to_string(),
                transform: RuleTransform::default(),
                endpoint: None,
                filter: None,
                event_handler: None,
                sources: None,
            },
            subscription_processed: None,
            workflow_callback: None,
        };

        // Act
        let result = thunder_broker.prepare_request(&broker_request);

        // Assert
        assert!(result.is_ok());
        let requests = result.unwrap();
        assert!(!requests.is_empty());
        // Additional assertions can be made on the content of the requests
    }

    // Additional tests can be added to cover other functionalities, error conditions,
    // and edge cases. Ensure that all public methods are covered.
}